package io.kiki.stack.netty.channelManager;


import io.kiki.stack.netty.invoker.Future;
import io.kiki.stack.netty.util.ConcurrentHashSet;
import io.kiki.stack.netty.util.InetSocketAddressUtil;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.util.AttributeKey;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Setter
@Getter
public class Channel {

    public static final AttributeKey<Channel> CHANNEL = AttributeKey.valueOf("channel");

    public static final AttributeKey<Integer> HEARTBEAT_COUNT = AttributeKey.valueOf("heartbeatCount");

    public static final AttributeKey<Boolean> HEARTBEAT_SWITCH = AttributeKey.valueOf("heartbeatSwitch");


    private final ConcurrentHashMap<Integer, Future> futureIdToFutureMap = new ConcurrentHashMap<Integer, Future>(4);
    private final ConcurrentHashMap<Integer/* id */, String/* poolKey */> id2PoolKey = new ConcurrentHashMap<Integer, String>(256);
    private final ConcurrentHashMap<String/* attr key*/, Object /*attr value*/> attributes = new ConcurrentHashMap<String, Object>();
    /**
     * the reference count used for this Channel. If equals 2, it means this Channel has been referenced 2 times
     */
    private final AtomicInteger referenceCount = new AtomicInteger();
    private final io.netty.channel.Channel channel;
    private final Set<String> poolKeys = new ConcurrentHashSet<String>();
    private final AtomicBoolean closed = new AtomicBoolean(false);


    private Url url;

    public Channel(io.netty.channel.Channel channel) {
        this.channel = channel;
        this.channel.attr(CHANNEL).set(this);
    }


    public Channel(io.netty.channel.Channel channel, Url url) {
        this(channel);
        this.url = url;
        //
        this.poolKeys.add(url.getId());
        //
        this.channel.attr(HEARTBEAT_SWITCH).set(true);
        this.channel.attr(HEARTBEAT_COUNT).set(0);

    }

    /**
     * to check whether the Channel is fine to use
     *
     * @return true if Channel is fine
     */
    public boolean isFine() {
        return this.channel != null && this.channel.isActive() && !closed.get();
    }


    public void increaseReferenceCount() {
        this.referenceCount.getAndIncrement();
    }

    public void decreaseReferenceCount() {
        this.referenceCount.getAndDecrement();
    }


    public boolean noReference() {
        return this.referenceCount.get() == 0;
    }


    public InetSocketAddress getRemoteAddress() {
        return (InetSocketAddress) this.channel.remoteAddress();
    }


    public String getRemoteIp() {
        return InetSocketAddressUtil.parseRemoteIp(this.channel);
    }

    public int getRemotePort() {
        return InetSocketAddressUtil.parseRemotePort(this.channel);
    }

    public InetSocketAddress getLocalAddress() {
        return (InetSocketAddress) this.channel.localAddress();
    }

    public String getLocalIp() {
        return InetSocketAddressUtil.parseLocalIp(this.channel);
    }

    public int getLocalPort() {
        return InetSocketAddressUtil.parseLocalPort(this.channel);
    }


    public Future getFuture(int futureId) {
        return this.futureIdToFutureMap.get(futureId);
    }


    public Future addFuture(Future future) {
        return this.futureIdToFutureMap.putIfAbsent(future.id(), future);
    }

    public Future removeFuture(int futureId) {
        return this.futureIdToFutureMap.remove(futureId);
    }


    public void onClose() {
        Iterator<Entry<Integer, Future>> iterator = futureIdToFutureMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<Integer, Future> entry = iterator.next();
            iterator.remove();
            Future future = entry.getValue();
            if (future != null) {
                future.putResponse(future.createChannelClosedResponse(this.getRemoteAddress()));
                future.cancelTimeout();
                future.tryAsyncExecuteInvokeCallbackAbnormally();
            }
        }
    }


    public void close() {
        if (closed.compareAndSet(false, true)) {
            try {
                if (this.getChannel() != null) {
                    this.getChannel().close().addListener(new ChannelFutureListener() {

                        @Override
                        public void operationComplete(ChannelFuture channelFuture) throws Exception {
                            if (log.isInfoEnabled()) {
                                log.info("Close the channel to remote address={}, result={}, cause={}", InetSocketAddressUtil.parseRemoteAddress(Channel.this.getChannel()), channelFuture.isSuccess(), channelFuture.cause());
                            }
                        }

                    });
                }
            } catch (Exception e) {
                log.warn("Exception caught when closing channel {}", InetSocketAddressUtil.parseRemoteAddress(Channel.this.getChannel()), e);
            }
        }
    }


    public boolean isIdToFutureMapIsEmpty() {
        return futureIdToFutureMap.isEmpty();
    }


    public void addPoolKey(String poolKey) {
        poolKeys.add(poolKey);
    }


    public Set<String> getPoolKeys() {
        return new HashSet<String>(poolKeys);
    }

    public void removePoolKey(String poolKey) {
        poolKeys.remove(poolKey);
    }


    public void addIdPoolKeyMapping(Integer id, String poolKey) {
        this.id2PoolKey.put(id, poolKey);
    }


    public String removeIdPoolKeyMapping(Integer id) {
        return this.id2PoolKey.remove(id);
    }


    public void setAttribute(String key, Object value) {
        attributes.put(key, value);
    }


    public Object setAttributeIfAbsent(String key, Object value) {
        return attributes.putIfAbsent(key, value);
    }


    public void removeAttribute(String key) {
        attributes.remove(key);
    }


    public Object getAttribute(String key) {
        return attributes.get(key);
    }


    public void clearAttributes() {
        attributes.clear();
    }


}
